/**
 * Copyright 2009-2010 by The Regents of the University of California
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * you may obtain a copy of the License from
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package edu.ucr.cs.dblab.acestore.base;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;

/**
 * @author jarodwen
 * @created Jan 9, 2011
 */
public class ACEContext {
	
	public final int pageSize;
	
	public final int bufferSize;
	
	private StorageManager storageManager;
	
	private int bufferRead, bufferWrite, swapCount;
	
	private ICacheAlgorithm cacheAlgorithm;
	
	/**
	 * The buffer cache.
	 */
	private Map<ACEPageKey, ACEPage> bufferCache;
	
	public ACEContext(int pageSize, int bufferSize, ICacheAlgorithm cacheAlgorithm) throws IOException{
		this.pageSize = pageSize;
		this.bufferSize = bufferSize;
		this.storageManager = new StorageManager(pageSize);
		this.bufferCache = new HashMap<ACEPageKey, ACEPage>();
		this.bufferRead = this.bufferWrite = this.swapCount = 0;
		this.cacheAlgorithm = cacheAlgorithm;
		this.cacheAlgorithm.setCache(bufferCache);
	}
	
	public ACEContext(int bufferSize, ICacheAlgorithm cacheAlgorithm, String dbLocation) throws IOException{
		this.bufferSize = bufferSize;
		this.storageManager = new StorageManager(dbLocation);
		this.pageSize = storageManager.getPageSize();
		this.bufferCache = new HashMap<ACEPageKey, ACEPage>();
		this.bufferRead = this.bufferWrite = this.swapCount = 0;
		this.cacheAlgorithm = cacheAlgorithm;
		this.cacheAlgorithm.setCache(bufferCache);
	}

	/**
	 * Read a page from the buffer cache. 
	 * 
	 * <ul>
	 * <li> If the page is in the buffer, it is directly loaded;
	 * <li> If the page is not in the buffer, it should be loaded from
	 * the storage. If during the loading there is no available space
	 * for the new page, swap is needed.
	 * </ul>
	 * @param idxID
	 * @param pageID
	 * @return
	 * @throws IOException
	 */
	public ByteBuffer readPage(int idxID, int pageID) throws IOException{
		bufferRead ++;
		ACEPageKey pageKey = new ACEPageKey(idxID, pageID);
		cacheAlgorithm.readPage(pageKey);
		if(bufferCache.containsKey(pageKey)){
			return bufferCache.get(pageKey).getPage();
		}else{
			ByteBuffer newPage = storageManager.load(idxID, pageID);
			if(bufferCache.size() >= bufferSize){
				swapPage();
				if(bufferCache.size() >= bufferSize){
					throw new IOException("Failed to swap any page out of the buffer.");
				}
			}
			bufferCache.put(pageKey, new ACEPage(newPage));
			return newPage;
		}
	}
	
	public ByteBuffer allocatePage(int idxID, int pageID) throws IOException{
		bufferRead ++;
		ACEPageKey pageKey = new ACEPageKey(idxID, pageID);
		cacheAlgorithm.readPage(pageKey);
		ByteBuffer newPage = ByteBuffer.wrap(new byte[pageSize]);
		if(bufferCache.size() >= bufferSize){
			swapPage();
			if(bufferCache.size() >= bufferSize){
				throw new IOException("Failed to swap any page out of the buffer.");
			}
		}
		bufferCache.put(pageKey, new ACEPage(newPage));
		return newPage;
	}
	
	public void writePage(int idxID, int pageID, ByteBuffer buf) throws IOException{
		bufferWrite ++;
		ACEPageKey pageKey = new ACEPageKey(idxID, pageID);
		cacheAlgorithm.writePage(pageKey);
		if(bufferCache.containsKey(pageKey)){
			bufferCache.put(pageKey, new ACEPage(buf, false, true));
		}else{
			if(bufferCache.size() >= bufferSize){
				swapPage();
				if(bufferCache.size() >= bufferSize){
					throw new IOException("Failed to swap any page out of the buffer.");
				}
			}
			bufferCache.put(pageKey, new ACEPage(buf, false, true));
		}
	}
	
	public void pinPage(int idxID, int pageID) throws IOException {
		ACEPageKey key = new ACEPageKey(idxID, pageID);
		if(bufferCache.containsKey(key)){
			bufferCache.get(key).pin();
		}else{
			throw new IOException("Failed to find the page to be pinned: (" + idxID + ":" + pageID + ")");
		}
	}
	
	public void unpinPage(int idxID, int pageID) throws IOException {
		ACEPageKey key = new ACEPageKey(idxID, pageID);
		if(bufferCache.containsKey(key)){
			bufferCache.get(key).unpin();
		}else{
			throw new IOException("Failed to find the page to be unpinned: (" + idxID + ":" + pageID + ")");
		}
	}
	
	/**
	 * Swap a page out of the buffer cache.
	 * 
	 * @throws IOException
	 */
	public void swapPage() throws IOException {
		this.swapCount ++;
		ACEPageKey key = cacheAlgorithm.swapPage();
		if(bufferCache.get(key).isDirty()){
			// Write the dirty page back to the storage
			storageManager.save(key.getFileID(), key.getPageID(), bufferCache.get(key).getPage());
		}
		// Remote the bytebuffer from the buffer cache
		bufferCache.remove(key);
	}
	
	/**
	 * Allocate a new file for a new index.
	 * 
	 * @return	The file id.
	 */
	public int allocateIndex(){
		return storageManager.allocateFile();
	}
	
	public void deallocateIndex(int idxID) throws IOException {
	    storageManager.deallocateFile(idxID);
	}
	
	public void flushBuffer() throws IOException{
		for(ACEPageKey key : bufferCache.keySet()){
			if(bufferCache.get(key).isDirty()){
				// Write the dirty page back to the storage
				storageManager.save(key.getFileID(), key.getPageID(), bufferCache.get(key).getPage());
			}
		}
		bufferCache.clear();
	}
	
	public String getIOStat(){
		return this.storageManager.getIOStat() + "Buffer Read\t" + this.bufferRead + "\nBuffer Write\t" + this.bufferWrite + "\nSwap Count\t" + this.swapCount;
	}
	
	public String getLocation(){
		return storageManager.dbLocation;
	}
	
	public String getIndexList(){
		return storageManager.listMeta();
	}
	
	public void close() throws IOException{
		this.flushBuffer();
		this.storageManager.close();
		System.out.println("ACE Context is closed now: " + getIOStat());
	}
}
